package com.dmp.storm.bolt.kafka;

import backtype.storm.Config;
import backtype.storm.Constants;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

import com.dmp.logger.ILogger;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static com.dmp.storm.base.BaseMapKey.MAPKEY_RESULTMAP_BOLTMSG;
import static com.dmp.storm.base.BaseMapKey.MAPKEY_TUPLEBOLT_RESULTMAP;

public abstract class AbstractBatchTickBolt extends BaseRichBolt {
	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;
	private String keyword;
    public String getBatchSize() {
		return batchSize;
	}

	protected OutputCollector collector;
    protected ILogger logger;
    
	private String rotateTime = "1";
	private String batchSize = "500";
	private long lastTime = 0L;
	public OutputCollector getCollector() {
		return collector;
	}

	public void setCollector(OutputCollector collector) {
		this.collector = collector;
	}

	public String getRotateTime() {
		return rotateTime;
	}

	public void setRotateTime(String rotateTime) {
		this.rotateTime = rotateTime;
	}

	public long getLastTime() {
		return lastTime;
	}

	public void setLastTime(long lastTime) {
		this.lastTime = lastTime;
	}

	public String getKeyword() {
		return keyword;
	}

	public void setBatchSize(String batchSize) {
		this.batchSize = batchSize;
	}
	
	private List<Map<String, Object>> tupleList = new ArrayList<Map<String, Object>>();
	
	@Override
	public final Map<String, Object> getComponentConfiguration() {
		Map<String, Object> conf = new HashMap<String, Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, Long.parseLong(rotateTime));
		
		return conf;
	}
	@SuppressWarnings({"rawtypes", "unchecked"})
    @Override
    public void execute(Tuple tuple) {
        try {
        	Map<String, Object> resultMap = null;
        	boolean isProcess = false;
        	
        	if(tuple.getSourceComponent().equals(Constants.SYSTEM_COMPONENT_ID) 
        			&& tuple.getSourceStreamId().equals(Constants.METRICS_TICK_STREAM_ID)){
        		if(!tupleList.isEmpty()){
        			long time = System.currentTimeMillis();
            		if(time - lastTime >= Integer.parseInt(rotateTime) * 1000){
            			isProcess = true;
            		}
        		}
        		
        	} else {
        		if(tuple.contains(MAPKEY_TUPLEBOLT_RESULTMAP)){
        			Map<String, Object> map = (Map<String, Object>) tuple.getValueByField(MAPKEY_TUPLEBOLT_RESULTMAP);
            		tupleList.add(map);
        		}
        		if(tuple.contains(MAPKEY_RESULTMAP_BOLTMSG)){
                
        		}
        		
        		
        		
        		
        		if(tupleList.size() >= Integer.parseInt(batchSize)) {
        			isProcess = true;
        		}
        	}
        	
        	if(isProcess){
        		resultMap = process(tupleList);
        		tupleList.clear();
        		lastTime = System.currentTimeMillis();
        	}
            
            if (resultMap != null) {
                collector.emit(tuple, new Values(resultMap, keyword));
            }
            
            collector.ack(tuple);
        } catch (Exception e) {
        	e.printStackTrace();
            collector.fail(tuple);
        }
    }

    @SuppressWarnings("rawtypes")
    @Override
    public void prepare(Map arg0, TopologyContext arg1, OutputCollector collector) {
        this.collector = collector;
        prepare();
    }
    
    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields(MAPKEY_TUPLEBOLT_RESULTMAP, "keyword"));
    }

    public abstract Map<String, Object> process(List<Map<String, Object>> tupleList);

    public void setKeyword(String keyword) {
        this.keyword = keyword;
    }

    public Object getMapBoltMsg(Map<String, Object> map) {
        return map.get(MAPKEY_RESULTMAP_BOLTMSG);
    }

	

	public void prepare() {
	    //logger = new LocationLogger();
	}
}
